package com.xiguaji.tiktok.task.music.service;

import com.gugee.tiktok.common.model.dal.blogger.BloggerLogRepository;
import com.gugee.tiktok.common.model.dal.blogger.BloggerTypeRepository;
import com.gugee.tiktok.common.model.dal.music.MusicRepository;
import com.gugee.tiktok.common.model.dto.DistributionRecord;
import com.gugee.tiktok.common.model.mysql.BloggerLog;
import com.gugee.tiktok.common.model.mysql.BloggerType;
import com.gugee.tiktok.data.common.msg.MusicMessage;
import com.xiguaji.boot.toolkit.JsonUtil;
import com.xiguaji.tiktok.task.common.config.AppConfig;
import com.xiguaji.tiktok.task.common.job.TaskParam;
import com.xiguaji.tiktok.task.common.producer.AbstractProducerHandler;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.apache.lucene.search.join.ScoreMode;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: luolifeng
 * Date: 2020-10-15 15:16
 */
@Service
@Slf4j
public class MusicService extends AbstractProducerHandler<MusicMessage> {
    @Autowired
    @Qualifier("kafkaTemplateMusic")
    private KafkaTemplate<Long, MusicMessage> kafkaTemplateMusic;


    public static String MUSIC_INDEX = "music_idx";

    @Autowired
    BloggerLogRepository bloggerLogRepository;

    @Autowired
    MusicRepository musicRepository;

    @Autowired
    RedisTemplate<String,String> redisTemplate;

    @Autowired
    RestHighLevelClient client;

    @Autowired
    BloggerTypeRepository bloggerTypeRepository;

    List<BloggerType> bloggerTypes = new ArrayList<>();
    @Autowired
    AppConfig appConfig;

    @PostConstruct
    public void initConstruct(){
        this.bloggerTypes.addAll(bloggerTypeRepository.getBloggerTypeByIsDisplay(true));
//        scheduleTemplate.attach(topBloggerMain, forkJoinPool);
    }

    @Override
    public List<MusicMessage> findMessageObject(BloggerLog bloggerLog, TaskParam taskParam) {
        return musicRepository.getMusicsTopOrderByScore((bloggerLog.getCurrentBloggerId()-1)*taskParam.getRecords(),taskParam.getRecords());

    }

    @Override
    public void sendMsg(List<MusicMessage> messages,String topic) {
        Set s=new HashSet<>(messages.size());
        messages.forEach(msg->{
            if(!s.contains(msg.getMid())){
                s.add(msg.getMid());
                msg.setRetryTimes(appConfig.getRetryMusicTimes());
                msg.setRetryLimitTimes(appConfig.getRetryMusicTimes());
                kafkaTemplateMusic.send(topic,msg);
            }

        });
        XxlJobLogger.log("投递主题={},投递消息数={}", topic,s.size());
        log.info("投递主题={},投递消息数={}", topic,s.size());
        s.clear();
    }

    public void setDistributionCash(){
        Set<Long> midList = musicRepository.getMusicsTopOrderByUserGrowth(0,1);
        bloggerTypes.parallelStream().forEach(bloggerType -> midList.addAll(getMusicsByCategory(bloggerType.getId())));
        midList.parallelStream().forEach( mid ->{
            setRegionDistribution(mid);
            setIndustryDistribution(mid);
        });
    }


    private void setRegionDistribution(long musicId){
        List<DistributionRecord> distributionRecords;
        long beforeDateTime = LocalDateTime.now().minusDays(90).atZone(ZoneId.systemDefault()).toEpochSecond();
        List<Map<String,Object>> mapList = musicRepository.getRegionDistribution(musicId,beforeDateTime);
        distributionRecords =  getDistributionRecords(mapList);
        distributionRecords.removeIf( x -> StringUtils.isEmpty(x.getDescription()));
        redisTemplate.opsForValue().set("dis_music_region_"+musicId, JsonUtil.parseJson(distributionRecords),7, TimeUnit.DAYS);
    }

    private void setIndustryDistribution(long musicId){
        List<DistributionRecord> distributionRecords;
        long beforeDateTime = LocalDateTime.now().minusDays(90).atZone(ZoneId.systemDefault()).toEpochSecond();
        List<Map<String,Object>> mapList = musicRepository.getIndustryDistribution(musicId,beforeDateTime);
        distributionRecords =  getDistributionRecords(mapList);
        distributionRecords.removeIf( x -> StringUtils.isEmpty(x.getDescription()));
        redisTemplate.opsForValue().set("dis_music_industry_"+musicId,JsonUtil.parseJson(distributionRecords),7, TimeUnit.DAYS);
    }


    private List<DistributionRecord> getDistributionRecords(List<Map<String, Object>> mapList) {
        List<DistributionRecord> distributionRecords = new ArrayList<>();
        long sum = mapList.stream().mapToLong(m -> (long) m.get("count")).sum();
        int otherCount = 0;

        for (int i = 0; i < mapList.size(); i++) {
            if(i < 9){
                DistributionRecord distributionRecord = new DistributionRecord();
                distributionRecord.setDescription(mapList.get(i).get("name") == null ? "" : mapList.get(i).get("name").toString());

                BigDecimal bigDecimal = new BigDecimal((long)mapList.get(i).get("count"));
                double percent =bigDecimal.divide(new BigDecimal(sum),3, RoundingMode.FLOOR).multiply(new BigDecimal(100)).setScale(2).doubleValue();
                if(percent <= 0){
                    continue;
                }
                distributionRecord.setPercent(percent);
                distributionRecords.add(distributionRecord);
            }else{
                otherCount += (long)mapList.get(i).get("count");
            }
        }

        if (otherCount > 0) {
            DistributionRecord distributionRecord = new DistributionRecord();
            distributionRecord.setDescription("Others");
            BigDecimal bigDecimal = new BigDecimal(otherCount);
            double percent =bigDecimal.divide(new BigDecimal(sum),3, RoundingMode.FLOOR).multiply(new BigDecimal(100)).setScale(2).doubleValue();
            distributionRecord.setPercent(percent);
            distributionRecords.add(distributionRecord);
        }

        distributionRecords.sort((o1,o2) ->{
            if(o1.getPercent() > o2.getPercent()){
                return -1;
            }else if(o1.getPercent() < o2.getPercent()){
                return 1;

            }
            return 0;
        });
        return distributionRecords;
    }


    private List<Long> getMusicsByCategory(long categoryId){

        List<Long> midList = new ArrayList<>();
        SearchRequest searchRequest = new SearchRequest(MusicService.MUSIC_INDEX);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.nestedQuery("categories",QueryBuilders.matchQuery("categories.id",categoryId), ScoreMode.Total));
        searchSourceBuilder.size(100);
        searchSourceBuilder.sort("userGrowth", SortOrder.DESC);
        searchRequest.source(searchSourceBuilder);
        SearchResponse searchResponse = null;
        try {
            searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }

        val hits = searchResponse.getHits().getHits();

        for (SearchHit hit : hits) {
            Map result = JsonUtil.jsonParse(Map.class,hit.getSourceAsString());
            midList.add(Long.parseLong(result.get("mid").toString()));
        }

        return midList;
    }
}
